package com._51doit.spark06

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object MovieTest02 {


  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
      .setMaster("local[*]")
      .setAppName(this.getClass.getSimpleName)

    val sc: SparkContext = new SparkContext(conf)

    val rateDat: RDD[String] = sc.textFile("D:\\07spark\\spark-day06\\作业题\\moviedata\\ratings.dat")
    val mvDat: RDD[String] = sc.textFile("D:\\07spark\\spark-day06\\作业题\\moviedata\\movies.dat")
    val usrDat: RDD[String] = sc.textFile("D:\\07spark\\spark-day06\\作业题\\moviedata\\users.dat")

    val res0: RDD[(String, String, Int)] = rateDat.map(rd => {
      val rateres1: Array[String] = rd.split("::")

      val uid: String = rateres1(0)
      val mid: String = rateres1(1)
      val rates: Int = rateres1(2).toInt
      //用户id  电影id  电影评分
      (uid, mid, rates)
    })

    //预处理的电影数据
            //用户id，电影id
    val resumd: RDD[(String, String)] = res0.map(rp=>(rp._1,rp._2))


    val res1: RDD[(String, String, String)] = usrDat.map(um => {
      val arrs: Array[String] = um.split("::")
      val uid: String = arrs(0)
      val gander: String = arrs(1)
      val age: String = arrs(2)

      (uid, gander,age)
    })

//数据过滤

    val res3: RDD[(String, String, String)] = res1.filter(rf=>(rf._2.equals("M")&&rf._3=="18"))
    //res3.foreach(println)

    val reswc: RDD[(String, Int)] = res3.map(ra=>(ra._1,1))

    //关联
    val res4: RDD[(String, (String, Int))] = resumd.join(reswc)

    val res5: RDD[(String, Int)] = res4.map(_._2).reduceByKey(_+_)

    val res6: Array[(String, Int)] = res5.sortBy(-_._2).take(10)

    //需求2结果
//res6.foreach(println)




//需求3

    val fres: RDD[(String, String, String)] = res1.filter(mf=>mf._2.equals("F"))
    val ffres: RDD[(String, Int)] = fres.map(fp=>(fp._1,1))
    //女生看过的所有电影
    val res7: RDD[(String, (String, Int))] = resumd.join(ffres)

    val res8: RDD[(String, Int)] = res7.map(p=>p._2).reduceByKey(_+_)


    val resmv: RDD[(String, String)] = mvDat.map(um => {
      val marr: Array[String] = um.split("::")
      val uid: String = marr(0)
      val mvname: String = marr(1)

      (uid, mvname)
    })
    val res9: RDD[(String, (String, Int))] = resmv.join(res8)

    val res10: Array[(String, Int)] = res9.map(a=>a._2).sortBy(-_._2).take(10)

    //res10.foreach(println)

    val resfinly: Array[String] = res10.map(_._1)

//需求3结果
    resfinly.foreach(println)













  }



}
